package com.shujia.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo4MapPartitions {
  def main(args: Array[String]): Unit = {
    /**
     * MapPartitions：转换算子
     *
     */
    // 构建Spark上下文环境
    val conf: SparkConf = new SparkConf()
    conf.setAppName("Demo4MapPartitions")
    conf.setMaster("local")
    val sc: SparkContext = new SparkContext(conf)


    val lineRDD: RDD[String] = sc.textFile("Spark/data/words")

    // 对每一个分区的数据进行处理
    // 适用于在算子内部需要跟外部数据源建立连接（一般创建连接是为了获取数据）的情况
    // 通过mapPartitions这种方式可以减少连接创建的次数，提高运行效率
    lineRDD.mapPartitions((iter: Iterator[String]) => {
      println("map partitions")
      iter.flatMap(line => line.split(","))
    }).foreach(println)

    lineRDD.mapPartitionsWithIndex((index, iter) => {
      println("当前的分区索引：" + index)
      iter.flatMap(line => line.split(","))
    }).foreach(println)


    // 对每一条数据进行处理，假设数据有N条
    // 如果需要在map中例如去请求MySQL的数据，那么会与MySQL建立N次连接
    // 会导致运行效率较低，甚至导致MySQL建立的连接数达到上限，出现性能问题
    lineRDD.map(line => {
      println("map")
      line.split(",")
    }).foreach(println)
  }

}
